package cn.dmp.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

object Tags4Business extends Tags{

  /**
    * 打标签的方法定义。参数是可变的，返回一个Map[String, Int]类型的
    *
    * @param args
    * @return
    */
  override def makeTags(args: Any*): Map[String, Int] = {
    var map = Map[String ,Int]()
    val row = args(0).asInstanceOf[Row]
    val jedis = args(1).asInstanceOf[Jedis]


    val lat = row.getAs[String]("lat")
    val longs = row.getAs[String]("long")

    if (StringUtils.isNotEmpty(lat) && StringUtils.isNotEmpty(longs)){
      val lat2 = lat.toDouble
      val lon2 = longs.toDouble

      if (lat2 > 3 && lat2 < 54 && lon2 > 73 && lon2 < 136){
        val geoHashCode = GeoHash.withCharacterPrecision(lat2, lon2, 8).toBase32
        val business = jedis.get(geoHashCode)
        if(StringUtils.isNotEmpty(business)){
          business.split(",").foreach(bs => map+="BS"+bs -> 1)
        }
      }
    }
    map
  }


}
